package com.atguigu.fink.chapter02;

import com.atguigu.fink.bean.WaterSensor;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;

/**
 * @Author lzc
 * @Date 2022/11/28 09:06
 */
public class Flink02_Table {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        DataStreamSource<WaterSensor> stream = env.fromElements(
            new WaterSensor("hello hello atguigu", 1L, 10),
            new WaterSensor("abc def", 2L, 10),
            new WaterSensor("aa hello a", 3L, 20),
            new WaterSensor("abc abc aaa", 4L, 30),
            new WaterSensor(null, 4L, 30)
        );
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        Table table = tEnv.fromDataStream(stream);
        tEnv.createTemporaryView("sensor", table);
        // 1. table api 中使用
        // 1.1 内联用法
        /*table
            //                        .joinLateral(call(MySplit.class, $("id")))
            .leftOuterJoinLateral(call(MySplit.class, $("id")))
            .select($("id"), $("word"), $("len"))
            .execute()
            .print();*/
        // 1.2. 先注册再使用
        // 2. 在sql 中使用
        tEnv.createTemporaryFunction("my_split", MySplit.class);
        // 内连接:
        // select .. from a join b on a.id=b.id
        // select .. from a, b where a.id=b.id
        /*tEnv.sqlQuery("select " +
                          "id, " +
                          "word, " +
                          "len " +
                          "from sensor " +
                          "join lateral table(my_split(id)) on true").execute().print();*/
    
        /*tEnv.sqlQuery("select " +
                          "id, " +
                          "word, " +
                          "len " +
                          "from sensor " +
                          ", lateral table(my_split(id))").execute().print();*/
    
        /*tEnv.sqlQuery("select " +
                          "id, " +
                          "word, " +
                          "len " +
                          "from sensor " +
                          "left join lateral table(my_split(id)) on true").execute().print();*/
    
    
        tEnv.sqlQuery("select " +
                          "id, " +
                          "abc.w, " +
                          "l " +
                          "from sensor " +
                          "left join lateral table(my_split(id)) as abc(w, l) on true").execute().print();
    
    
    }
    
    /*@FunctionHint(output = @DataTypeHint("row<word string, len int>"))
    public static class MySplit extends TableFunction<Row> {
        *//*
        约定方法:
        返回值: 必须是 void
        方法名: 必须是 eval
        参数列表: 根据事件情况
         *//*
        public void eval(String s) {
            if (s == null || s.contains("atguigu")) {
                return;
            }
            
            
            String[] words = s.split(" ");
            for (String word : words) {
                collect(Row.of(word, word.length()));
            }
            
        }
    }*/
    
    @NoArgsConstructor
    @AllArgsConstructor
    @Data
    public static class WordLen{
        private String word;
        private Integer len;
    }
    
    
    public static class MySplit extends TableFunction<WordLen> {
        
        public void eval(String s) {
            if (s == null || s.contains("atguigu")) {
                return;
            }
            
            String[] words = s.split(" ");
            for (String word : words) {
                collect(new WordLen(word, word.length()));
            }
            
        }
    }
}
/*
"hello hello atguigu"
    hello    5
    hello    5
    atguigu  7
    
"hello hello atguigu"       hello    5
"hello hello atguigu"       hello    5
"hello hello atguigu"       atguigu  7

"abc def"
"abc def"    abc 3
"abc def"   def 3
 */